package com.ruding.disruptor.demo;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadFactory;

import com.lmax.disruptor.BlockingWaitStrategy;
import com.lmax.disruptor.EventFactory;
import com.lmax.disruptor.EventHandler;
import com.lmax.disruptor.RingBuffer;
import com.lmax.disruptor.WaitStrategy;
import com.lmax.disruptor.WorkHandler;
import com.lmax.disruptor.YieldingWaitStrategy;
import com.lmax.disruptor.dsl.Disruptor;
import com.lmax.disruptor.dsl.ProducerType;

/**
 * Hello world!
 *
 */
public class Main {
    @SuppressWarnings("unchecked")
	public static void main( String[] args ){
    	/**
    	 * 指定等待策略
		BlockingWaitStrategy 是最低效的策略，但其对CPU的消耗最小并且在各种不同部署环境中能提供更加一致的性能表现；
		SleepingWaitStrategy 的性能表现跟 BlockingWaitStrategy 差不多，对 CPU 的消耗也类似，但其对生产者线程的影响最小，适合用于异步日志类似的场景；
		YieldingWaitStrategy 的性能是最好的，适合用于低延迟的系统。在要求极高性能且事件处理线数小于 CPU 逻辑核心数的场景中，推荐使用此策略；例如，CPU开启超线程的特性。
    	 */
    	EventFactory<LongEvent> eventFactory = new LongEventFactory();
    	ExecutorService executor = Executors.newFixedThreadPool(10);
    	int ringBufferSize = 1024 * 1024; // RingBuffer 大小，必须是 2 的 N 次方；
    	@SuppressWarnings("deprecation")
		Disruptor<LongEvent> disruptor = new Disruptor<LongEvent>(eventFactory,
    	                ringBufferSize, 
    	                executor,
    	                ProducerType.SINGLE,
    	                new BlockingWaitStrategy()
    	                );
    	        
    	/*EventHandler<LongEvent> eventHandler1 = new LongEventHandler(1);
    	EventHandler<LongEvent> eventHandler2 = new LongEventHandler(2);
    	//多消费者重复消费
    	disruptor.handleEventsWith(eventHandler1,eventHandler2);*/
    	 /*
         * 该方法传入的消费者需要实现WorkHandler接口，方法的内部实现是：先创建WorkPool，然后封装WorkPool为EventHandlerPool返回。
         * 消费者1、2对于消息的消费有时有竞争，保证同一消息只能有一个消费者消费
         */
    	WorkHandler<LongEvent> workHandler1 = new LongEventHandler(1);
    	WorkHandler<LongEvent> workHandler2 = new LongEventHandler(2);
        disruptor.handleEventsWithWorkerPool(workHandler1, workHandler2);
    	disruptor.start();
    	
    	long start = System.currentTimeMillis();
    	RingBuffer<LongEvent> ringBuffer = disruptor.getRingBuffer();
    	LongEventProducer producer = new LongEventProducer(ringBuffer);
    	System.out.println(ringBuffer.remainingCapacity());
    	System.out.println(ringBuffer.getBufferSize());
    	for (int i = 0; i < 20; i++) {
    		producer.onData(i);
    		
    		
		}
    	System.out.println(ringBuffer.getMinimumGatingSequence());
    	System.out.println("耗时s：" + (System.currentTimeMillis()-start));
    	System.out.println(ringBuffer.getMinimumGatingSequence());
    	disruptor.shutdown();//关闭 disruptor，方法会堵塞，直至所有的事件都得到处理；
    	System.out.println(ringBuffer.remainingCapacity());
    	System.out.println(ringBuffer.getMinimumGatingSequence());
    	System.out.println("耗时：" + (System.currentTimeMillis()-start));
    	executor.shutdown();//关闭 disruptor 使用的线程池；如果需要的话，必须手动关闭， disruptor 在 shutdown 时不会自动关闭；
    }

}
